package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author : ranzlupup
 * @date : 2023/3/4 17:10
 */
object StreamTest {
    def main(args: Array[String]): Unit = {

        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
        val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

        val linesStream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)

        val wordStreams: DStream[String] = linesStream.flatMap(_.split(" "))
        val wordAndOneStreams: DStream[(String, Int)] = wordStreams.map(word => (word, 1))
        val wordAndCountStreams: DStream[(String, Int)] = wordAndOneStreams.reduceByKey(_ + _)
        wordAndCountStreams.print()

        ssc.start()
        ssc.awaitTermination()
    }
}
